package com.guanzi.big_data_mq_process.producer;

import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.guanzi.big_data_mq_process.constant.MessageConstant;
import com.guanzi.big_data_mq_process.model.OrderCreateContext;
import com.guanzi.big_data_mq_process.model.dto.OrdersDTO;
import com.guanzi.big_data_mq_process.model.entity.TransactionMessage;
import com.guanzi.big_data_mq_process.model.enums.LocalTransactionStateEnum;
import com.guanzi.big_data_mq_process.service.IOrderDetailsService;
import com.guanzi.big_data_mq_process.service.IOrdersService;
import com.guanzi.big_data_mq_process.service.ITransactionMessageService;
import com.guanzi.big_data_mq_process.service.IUsersService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.util.IdGenerator;

import java.util.concurrent.CompletableFuture;

/**
 * 生产者服务
 */
@Slf4j
@Service
public class OrderProducerService {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private IOrdersService ordersService;

    @Autowired
    private ITransactionMessageService transactionMessageService;

    /**
     * 创建订单事务消息
     *
     * @param dto
     */
    public void createOrderTransactionMessage(OrdersDTO dto) {
        String messageId = dto.getTransactionId();

        // TODO orderId是可以用雪花算法预先生成的，这里偷懒先查数据库下一个自增ID
        Long orderId = ordersService.getNextId();

        OrderCreateContext context = new OrderCreateContext(
                messageId,
                orderId,
                dto.getUserId(),
                dto.getTotalAmount(),
                dto.getDetails()
        );

        // 创建消息
        Message<OrderCreateContext> message = MessageBuilder
                .withPayload(context)
                .setHeader(RocketMQHeaders.TRANSACTION_ID, messageId)
                // 添加自定义TAG，给不同的消费者过滤消费
                // 发送到事务监听器后，这个自己设定的TAGS属性会丢失，但是其他的属性不会
                .setHeader(RocketMQHeaders.TAGS, MessageConstant.ORDER_TAG)
                // 添加自定义KEY
                .setHeader(RocketMQHeaders.KEYS, "order-key")
                .build();

        log.info("创建消息成功! 用户为: 【{}】, 订单号为: 【{}】 消息的header为: {}", dto.getUserId(), orderId, message.getHeaders());

        // 发送事务消息
        rocketMQTemplate.sendMessageInTransaction(MessageConstant.ORDER_TOPIC, message, context);
        log.info("发送事务消息成功，消息ID：{}", messageId);

        // 异步提交事务状态 经测试可以保证在信息塞入本地消息表之后，才来做这个更新
        // 因为事务半消息这里发送事务消息和Broker的接收响应是原子性的，即时就会立刻调用executeLocalTransaction方法
        CompletableFuture.runAsync(() -> {
            if (transactionMessageService.exists(Wrappers.<TransactionMessage>query().eq("message_id", messageId))) {
                transactionMessageService.updateTxState(messageId, LocalTransactionStateEnum.COMMITTED);
            }
        });
    }

}
